/*
 * Copyright 2012-2020 the original author or authors.
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 *      https://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */

package org.springframework.cloud.loadbalancer.core;

import java.util.ArrayList;
import java.util.List;
import java.util.Map;
import java.util.Random;
import java.util.concurrent.atomic.AtomicInteger;

import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.springframework.beans.factory.ObjectProvider;
import org.springframework.cloud.client.ServiceInstance;
import org.springframework.cloud.client.loadbalancer.DefaultResponse;
import org.springframework.cloud.client.loadbalancer.EmptyResponse;
import org.springframework.cloud.client.loadbalancer.Request;
import org.springframework.cloud.client.loadbalancer.Response;
import org.springframework.core.env.Environment;

import com.alibaba.cloud.nacos.NacosServiceInstance;
import com.ruoyi.common.core.utils.SpringUtils;

import breeze.framework.os.OS;
import breeze.framework.string.Str;
import reactor.core.publisher.Mono;

/**
 * A Round-Robin-based implementation of {@link ReactorServiceInstanceLoadBalancer}.
 *
 * @author Spencer Gibb
 * @author Olga Maciaszek-Sharma
 */
public class RoundRobinLoadBalancer implements ReactorServiceInstanceLoadBalancer {

	private static final Log log = LogFactory.getLog(RoundRobinLoadBalancer.class);

	final AtomicInteger position;

	final String serviceId;

	ObjectProvider<ServiceInstanceListSupplier> serviceInstanceListSupplierProvider;

	/**
	 * @param serviceInstanceListSupplierProvider a provider of
	 * {@link ServiceInstanceListSupplier} that will be used to get available instances
	 * @param serviceId id of the service for which to choose an instance
	 */
	public RoundRobinLoadBalancer(ObjectProvider<ServiceInstanceListSupplier> serviceInstanceListSupplierProvider,
			String serviceId) {
		this(serviceInstanceListSupplierProvider, serviceId, new Random().nextInt(1000));
	}

	/**
	 * @param serviceInstanceListSupplierProvider a provider of
	 * {@link ServiceInstanceListSupplier} that will be used to get available instances
	 * @param serviceId id of the service for which to choose an instance
	 * @param seedPosition Round Robin element position marker
	 */
	public RoundRobinLoadBalancer(ObjectProvider<ServiceInstanceListSupplier> serviceInstanceListSupplierProvider,
			String serviceId, int seedPosition) {
		this.serviceId = serviceId;
		this.serviceInstanceListSupplierProvider = serviceInstanceListSupplierProvider;
		this.position = new AtomicInteger(seedPosition);
	}

	@SuppressWarnings("rawtypes")
	@Override
	// see original
	// https://github.com/Netflix/ocelli/blob/master/ocelli-core/
	// src/main/java/netflix/ocelli/loadbalancer/RoundRobinLoadBalancer.java
	public Mono<Response<ServiceInstance>> choose(Request request) {
		ServiceInstanceListSupplier supplier = serviceInstanceListSupplierProvider
				.getIfAvailable(NoopServiceInstanceListSupplier::new);
		return supplier.get(request).next()
				.map(serviceInstances -> processInstanceResponse(supplier, serviceInstances));
	}

	private Response<ServiceInstance> processInstanceResponse(ServiceInstanceListSupplier supplier,
			List<ServiceInstance> serviceInstances) {
		Response<ServiceInstance> serviceInstanceResponse = getInstanceResponse(serviceInstances);
		if (supplier instanceof SelectedInstanceCallback && serviceInstanceResponse.hasServer()) {
			((SelectedInstanceCallback) supplier).selectedServiceInstance(serviceInstanceResponse.getServer());
		}
		return serviceInstanceResponse;
	}

	/**
	 * 为了测试环境和本机环境可以连接同一个nacos而临时写的ip判断.
	 * TODO 至少让本地的微服务可以连通线上的微服务.
	 * TODO 需要替换成高效的负载均衡策略,参考ribbon的七种实现.
	 */
	private Response<ServiceInstance> getInstanceResponse(List<ServiceInstance> instances) {
		if (instances.isEmpty()) {
			if (log.isWarnEnabled()) {
				log.warn("No servers available for service: " + serviceId);
			}
			return new EmptyResponse();
		}
		before(instances);
		ServiceInstance instance = null;
		String instanceType = null;
		try {
			
			List<String> local_ips = OS.getLocalIPList();


			/** 本机程序最新,访问速度最快,优先考虑 */
			List<ServiceInstance> instances_group_local = new ArrayList<>();
			
			/**
			 * 本地局域网可以互通
			 * TODO 生产内网很多无法互通,需要额外处理
			 */
			List<ServiceInstance> instances_group_lan = new ArrayList<>();
			
			/** 外网理论上都可以访问,但是外网的实例注册的地址可能是内网地址,需要在nacos进行配置. */
			List<ServiceInstance> instances_group_wan = new ArrayList<>();

			for (int i = 0; i < instances.size(); i++) {
				ServiceInstance serviceInstance = instances.get(i);
				String host = serviceInstance.getHost();
				if(local_ips.contains(host)) {
					instances_group_local.add(serviceInstance);
				}
				
				//默认ip前3位相同算是同一个局域网段.
				boolean isLan = false;
				for (int j = 0; j < local_ips.size(); j++) {
					String local_ip = local_ips.get(j);
					if(host.startsWith(local_ip.substring(0, Str.indexOf(local_ip, ".", 3)))) {
						isLan = true;
						break;
					}
				}
				if(isLan) {
					instances_group_lan.add(serviceInstance);
				}else {
					instances_group_wan.add(serviceInstance);
				}
			}
			
			
			
			//在每组实例中进行轮询
			int pos = Math.abs(this.position.incrementAndGet());
			if(instances_group_local.size() > 0) {
				instance = instances_group_local.get(pos % instances_group_local.size());
				instanceType = "local";
			}else if(instances_group_lan.size() > 0) {
				instance = instances_group_lan.get(pos % instances_group_lan.size());
				instanceType = "lan";
			}else if(instances_group_wan.size() > 0) {
				//lan->wan
				//通过读取yml文件实现.
				String lanToWan = SpringUtils.getBean(Environment.class).getProperty("platform.lanToWan");
				if(null != lanToWan) {
					String lan = lanToWan.split("-")[0];
					String wan = lanToWan.split("-")[1];
					for (int i = 0; i < instances_group_wan.size(); i++) {
						ServiceInstance instance_group_wan = instances_group_wan.get(i);
						if(instance_group_wan.getHost().equalsIgnoreCase(lan) && instance_group_wan.getClass() == NacosServiceInstance.class) {
							NacosServiceInstance nacosServiceInstance = (NacosServiceInstance) instance_group_wan;
							nacosServiceInstance.setHost(wan);
							instance = instance_group_wan;
							break;
						}
					}
				}else {
					//通过nacos服务列表配置元数据实现.
					for (int i = 0; i < instances_group_wan.size(); i++) {
						ServiceInstance instance_group_wan = instances_group_wan.get(i);
						try {
							Map<String, String> metadata = instance_group_wan.getMetadata();
							if(null != metadata && metadata.containsKey("lanToWan")) {
								lanToWan = metadata.get("lanToWan").toString().trim();
								String lan = lanToWan.split("-")[0];
								String wan = lanToWan.split("-")[1];
								if(instance_group_wan.getHost().equalsIgnoreCase(lan) && instance_group_wan.getClass() == NacosServiceInstance.class) {
									NacosServiceInstance nacosServiceInstance = (NacosServiceInstance) instance_group_wan;
									nacosServiceInstance.setHost(wan);
									instance = instance_group_wan;
									break;
								}
							}
						} catch (Exception e) {
							e.printStackTrace();
						}
					}
				}
				if(null == instance) {
					instance = instances_group_wan.get(pos % instances_group_wan.size());
				}
				instanceType = "wan";
			}else {
				//通过Ip分组没取到,尝试找到能够连通的微服务实例.
//				for (int i = 0; i < instances.size(); i++) {
//					instance = instances.get(i);
//					InetAddress address = InetAddress.getByName(instance.getHost());
//					if(address.isReachable(3000)) {
//						break;
//					}
//				}
				instanceType = "reachable";
			}
			log.debug("instance:[" + instanceType + ", " + instance.getServiceId() + ", " + instance.getHost() + ":" + instance.getPort() + "]");
//			System.out.println("local_ips : " + local_ips);
//			
//			System.out.print("instances_group_local : ");
//			instances_group_local.stream().forEach(ins -> {System.out.print(ins.getHost() + ",");});
//			System.out.println();
//			
//			System.out.print("instances_group_lan : ");
//			instances_group_lan.stream().forEach(ins -> {System.out.print(ins.getHost() + ",");});
//			System.out.println();
//			
//			System.out.print("instances_group_wan : ");
//			instances_group_wan.stream().forEach(ins -> {System.out.print(ins.getHost() + ",");});
//			System.out.println();
//			
//			System.out.print("instances : ");
//			instances.stream().forEach(ins -> {System.out.print(ins.getHost() + ",");});
//			System.out.println();
		} catch (Exception e) {
			e.printStackTrace();
		}
		
		return new DefaultResponse(instance);
	}

	//wan->lan
	private void before(List<ServiceInstance> instances) {
		for (int i = 0; i < instances.size(); i++) {
			ServiceInstance instance = instances.get(i);
			try {
				Map<String, String> metadata = instance.getMetadata();
				if(null != metadata && metadata.containsKey("lanToWan")) {
					String lanToWan = metadata.get("lanToWan").toString().trim();
					String lan = lanToWan.split("-")[0];
					String wan = lanToWan.split("-")[1];
					if(instance.getHost().equalsIgnoreCase(wan) && instance.getClass() == NacosServiceInstance.class) {
						NacosServiceInstance nacosServiceInstance = (NacosServiceInstance) instance;
						nacosServiceInstance.setHost(lan);
					}
				}
			} catch (Exception e) {
				System.err.println(e);
			}
		}
	}
}
